package rpc.netty.tcp;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.Environment;
import rpc.core.InitClose;
import rpc.utils.TypeSafe;

/*********
 *
 *
 *@author 197
 *@date 2020/6/19 13:49
 *
 **/
public class NettyServer implements InitClose {

  private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
  private final ConcurrentHashMap<String, Channel> conns = new ConcurrentHashMap<>();
  private final int maxPackageSize = 1000_000_000;
  private int coreThreads = 2;
  private int port;
  private boolean nativeNetty = false;
  private ServerBootstrap serverBootstrap;
  private EventLoopGroup bossGroup;
  private EventLoopGroup workerGroup;
  private Channel serverChannel;
  private int workerThreads = 2;
  private Environment environment;

  public NettyServer setPort(int port) {
    this.port = port;
    return this;
  }

  public NettyServer setEnvironment(Environment environment) {
    this.environment = environment;
    return this;
  }

  @Override
  public void init() {
    coreThreads = TypeSafe.anyToInt(environment.getProperty("rpc.netty.boss.threads"), 8);
    workerThreads = TypeSafe.anyToInt(environment.getProperty("rpc.netty.worker.threads"), 8);
    String osName = System.getProperty("os.name");
    String linux = "linux";
    if (nativeNetty && StringUtils.isNotBlank(osName) && osName.toLowerCase().contains(linux)) {
      nativeNetty = true;
    }
    ThreadFactory bossThreadFactory = new ThreadFactoryBuilder()
        .setNameFormat("rpc_nettyServer_boss-%d").build();
    ThreadFactory workThreadFactory = new ThreadFactoryBuilder()
        .setNameFormat("rpc_nettyServer_worker-%d").build();
    if (nativeNetty) {
      bossGroup = new EpollEventLoopGroup(coreThreads, bossThreadFactory);
      workerGroup = new EpollEventLoopGroup(workerThreads, workThreadFactory);
    } else {
      bossGroup = new NioEventLoopGroup(coreThreads, bossThreadFactory);
      workerGroup = new NioEventLoopGroup(workerThreads, workThreadFactory);
    }
    serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class);
    serverBootstrap.option(ChannelOption.SO_BACKLOG, 128);
    serverBootstrap.option(ChannelOption.SO_REUSEADDR, true);
    serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
    serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
    serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        int idleSeconds = 180;
        pipeline.addLast("frame-decoder",
            new RpcMessageDecoder(maxPackageSize, 4, 4, 0, 0));
        pipeline.addLast(new NettServeryHandler());
        pipeline.addLast("RpcMessageEncoder", new RpcMessageEncoder());
        pipeline.addLast("timeout", new IdleStateHandler(0, 0, idleSeconds));
      }
    });
  }

  @Override
  public void start() {
    InetSocketAddress addr = new InetSocketAddress(this.port);
    serverChannel = serverBootstrap.bind(addr).syncUninterruptibly().channel();
    logger.info("netty server started on host(*) port(" + port + ")");
  }

  @Override
  public void close() {
    if (workerGroup != null) {
      logger.info("stopping netty server port:{}", port);
      bossGroup.shutdownGracefully();
      bossGroup = null;
      ChannelGroup allChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
      allChannels.add(serverChannel);
      allChannels.addAll(conns.values());
      ChannelGroupFuture future = allChannels.close();
      future.awaitUninterruptibly();
      workerGroup.shutdownGracefully();
      workerGroup = null;
      logger.info("netty server stopped port:{}", port);
    }
  }
}